package com.xiaofu.live.user.provider.config;

import com.alibaba.fastjson2.JSON;
import com.xiaofu.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;
import com.xiaofu.live.user.dto.CacheAsyncDeleteCode;
import com.xiaofu.live.user.dto.UserCacheAsyncDeleteDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;

@Configuration
@Slf4j
public class RocketMQConsumerConfig implements InitializingBean {

    @Resource
    private RocketMQConsumerProperties consumerProperties;
    @Autowired
    private RocketMQConsumerProperties rocketMQConsumerProperties;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private UserProviderCacheKeyBuilder userProviderCacheKeyBuilder;

    public static final String CACHE_ASYNC_DELETE = "UserCacheAsyncDelete";

    @Override
    public void afterPropertiesSet() throws Exception {
        initConsumer();
    }

    private void initConsumer() throws MQClientException {
        // 初始化RocketMQ消费者
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr(consumerProperties.getNameSrv());
        defaultMQPushConsumer.setConsumerGroup(consumerProperties.getGroupName());
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        defaultMQPushConsumer.subscribe(CACHE_ASYNC_DELETE, "*");
        defaultMQPushConsumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                String msg = new String(msgs.get(0).getBody());
                log.info("userInfo is {}", msg);
                UserCacheAsyncDeleteDTO userCacheAsyncDeleteDTO = JSON.parseObject(msg, UserCacheAsyncDeleteDTO.class);
                String redisKey = "";
                Long userId = JSON.parseObject(userCacheAsyncDeleteDTO.getJson()).getLong("userId");
                if (CacheAsyncDeleteCode.USER_INFO_DELETE.getCode() == userCacheAsyncDeleteDTO.getCode()) {
                    redisKey = userProviderCacheKeyBuilder.buildUserInfoKey(userId);
                } else if (CacheAsyncDeleteCode.USER_TAG_DELETE.getCode() == userCacheAsyncDeleteDTO.getCode()) {
                    redisKey = userProviderCacheKeyBuilder.buildUserTagInfoKey(userId);
                }
                if (StringUtils.isNotBlank(redisKey)) {
                    redisTemplate.delete(redisKey);
                    log.info("延迟删除处理 userId is {}", userId);
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }

        });
        defaultMQPushConsumer.start();
        log.info("mq消费者启动成功,namesrv is {}", rocketMQConsumerProperties.getNameSrv());
    }
}
